package com.raylu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.raylu.realtime.app.template.DimAsyncFunction;
import com.raylu.realtime.bean.OrderDetail;
import com.raylu.realtime.bean.OrderInfo;
import com.raylu.realtime.bean.OrderWide;
import com.raylu.realtime.utils.*;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.time.Period;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2021/12/30.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
public class OrderWideApp2 {
    public static void main(String[] args) throws Exception {
        Properties load = PropertiesUtil.load("config.properties");
        //TODO 1. 准备环境变量
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2. 设置检查点
//        env.enableCheckpointing(5000L);
        //TODO 2.1. 设置检查点分解线对齐
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //TODO 2.2. 设置检查点超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
        //TODO 2.3. 设置当job cancel时是否保留检查点
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //TODO 2.4. 设置失败重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
        //TODO 2.5. 设置检查点存放位置
//        env.setStateBackend(new FsStateBackend(load.getProperty("order.wide.app.fsstatebackend.url")));
//        System.setProperty("HADOOP_USER_NAME", "raylu");
        //TODO 3.  从Kafka（DWD_ORDER_INFO）中读取OrderInfo数据
        KeyedStream<OrderInfo, String> orderInfoKeyedStream = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("order.wide.app.kafka.source-topic1"), load.getProperty("order.wide.app.kafka.group-id")))
                .map(new RichMapFunction<String, OrderInfo>() {
                    private SimpleDateFormat simpleDateFormat;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderInfo map(String value) throws Exception {
                        OrderInfo orderInfo = JSON.parseObject(value, OrderInfo.class);
                        orderInfo.setTs(simpleDateFormat.parse(orderInfo.getCreate_time()).getTime());
                        return orderInfo;
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                            @Override
                            public long extractTimestamp(OrderInfo element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })).keyBy(r -> r.getId());

        //TODO 4. 从Kafka（DWD_ORDER_DETAIL）中读取OrderDetail数据
        KeyedStream<OrderDetail, String> orderDetailKeyedStream = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("order.wide.app.kafka.source-topic2"), load.getProperty("order.wide.app.kafka.group-id")))
                .map(new RichMapFunction<String, OrderDetail>() {
                    private SimpleDateFormat simpleDateFormat;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderDetail map(String value) throws Exception {
                        OrderDetail orderDetail = JSON.parseObject(value, OrderDetail.class);
                        orderDetail.setTs(simpleDateFormat.parse(orderDetail.getCreate_time()).getTime());
                        return orderDetail;
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderDetail>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderDetail>() {
                            @Override
                            public long extractTimestamp(OrderDetail element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })).keyBy(r -> r.getOrder_id());

        //TODO 5. 通过Interval Join 将两条流进行关联，OrderInfo流为主流设置间隔读取OrderDetail流
        SingleOutputStreamOperator<OrderWide> orderWideStreamOperator = orderInfoKeyedStream
                .intervalJoin(orderDetailKeyedStream)
                .between(Time.seconds(-5L), Time.seconds(5L))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {

                    @Override
                    public void processElement(OrderInfo left, OrderDetail right, ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>.Context ctx, Collector<OrderWide> out) throws Exception {
                        OrderWide orderWide = new OrderWide(left, right);
                        out.collect(orderWide);
                    }
                });
        orderWideStreamOperator.print("5. ");
        //TODO 6. 通过map将OrderWide的数据进行维度补充
//        orderWideStreamOperator.map(new RichMapFunction<OrderWide, OrderWide>() {
//                    @Override
//                    public OrderWide map(OrderWide value) throws Exception {
//                        JSONObject jsonObject = DimUtil.qurayTableByPkWithCache(load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_user_info", Tuple2.of("id", value.getUser_id()));
//                        String login_name = jsonObject.getString("LOGIN_NAME");
//                        String name = jsonObject.getString("NAME");
//                        String birthday = jsonObject.getString("BIRTHDAY");
//                        String gender = jsonObject.getString("GENDER");
//                        value.setLogin_name(login_name);
//                        value.setName(name);
//                        value.setBirthday(birthday);
//                        value.setGender(gender);
//                        LocalDate birthdayDate = LocalDate.parse(birthday);
//                        Period between = Period.between(LocalDate.now(), birthdayDate);
//                        int age = between.getYears();
//                        value.setAge(age);
//                        return value;
//                    }
//                }).print();

//        //Test
//
//        AsyncDataStream.unorderedWait(orderWideStreamOperator,
//                new RichAsyncFunction<OrderWide, OrderWide>() {
//                    private ThreadPoolExecutor threadPoolExecutor;
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        super.open(parameters);
//                        threadPoolExecutor = ThreadUtil.getInstance();
//                    }
//
//                    @Override
//                    public void asyncInvoke(OrderWide input, ResultFuture<OrderWide> resultFuture) throws Exception {
//                        threadPoolExecutor.submit(new Runnable() {
//                            @Override
//                            public void run() {
//                                resultFuture.complete(Collections.singleton(input));
//                            }
//                        });
//
//                    }
//                },60L,TimeUnit.SECONDS).print("Test>>>");

        //TODO 6. 通过异步IO的方式，提高查询效率
        //TODO 6.1. 通过异步IO的方式关联用户维度数据
        SingleOutputStreamOperator<OrderWide> orderWideDSWithUser = AsyncDataStream.unorderedWait(orderWideStreamOperator,
                new DimAsyncFunction<OrderWide>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2<String, String>[] tuple2s = new Tuple2[1];
                        tuple2s[0] = Tuple2.of("user_id", "id");
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_user_info";
                    }

                    @Override
                    public void join(OrderWide value, JSONObject jsonObject) {
                        String login_name = jsonObject.getString("LOGIN_NAME");
                        String name = jsonObject.getString("NAME");
                        String birthday = jsonObject.getString("BIRTHDAY");
                        String gender = jsonObject.getString("GENDER");
                        value.setLogin_name(login_name);
                        value.setName(name);
                        value.setBirthday(birthday);
                        value.setGender(gender);
                        LocalDate birthdayDate = LocalDate.parse(birthday);
                        Period between = Period.between(birthdayDate,LocalDate.now());
                        int age = between.getYears();
                        value.setAge(age);
                    }


                }, 60 * 5, TimeUnit.SECONDS);


        //TODO 6.2. 通过异步IO的方式关联城市维度数据
        SingleOutputStreamOperator<OrderWide> orderWideDSWithProvince = AsyncDataStream.unorderedWait(orderWideDSWithUser,
                new DimAsyncFunction<OrderWide>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2<String, String>[] tuple2s = new Tuple2[1];
                        tuple2s[0] = Tuple2.of("province_id", "id");
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_base_province";
                    }

                    @Override
                    public void join(OrderWide value, JSONObject jsonObject) {
                        String province_name = jsonObject.getString("NAME");
                        String region_id = jsonObject.getString("REGION_ID");
                        String area_code = jsonObject.getString("AREA_CODE");
                        String iso_code = jsonObject.getString("ISO_CODE");
                        String iso_3166_2 = jsonObject.getString("ISO_3166_2");

                        value.setProvince_name(province_name);
                        value.setRegion_id(region_id);
                        value.setArea_code(area_code);
                        value.setIso_code(iso_code);
                        value.setIso_3166_2(iso_3166_2);

                    }


                }, 60 * 5, TimeUnit.SECONDS);

        //TODO 6.3. 通过异步IO的方式关联地区维度数据
        SingleOutputStreamOperator<OrderWide> orderWideDSWithRegion = AsyncDataStream.unorderedWait(orderWideDSWithProvince,
                new DimAsyncFunction<OrderWide>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2<String, String>[] tuple2s = new Tuple2[1];
                        tuple2s[0] = Tuple2.of("region_id", "id");
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_base_region";
                    }

                    @Override
                    public void join(OrderWide value, JSONObject jsonObject) {
                        String region_name = jsonObject.getString("REGION_NAME");
                        value.setRegion_name(region_name);
                    }

                }, 60 * 5, TimeUnit.SECONDS);

        //TODO 6.4. 通过异步IO的方式关联sku维度数据
        SingleOutputStreamOperator<OrderWide> orderWideDSWithSku = AsyncDataStream.unorderedWait(orderWideDSWithRegion,
                new DimAsyncFunction<OrderWide>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2<String, String>[] tuple2s = new Tuple2[1];
                        tuple2s[0] = Tuple2.of("sku_id", "id");
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_sku_info";
                    }

                    @Override
                    public void join(OrderWide value, JSONObject jsonObject) {
                        String spu_id = jsonObject.getString("SPU_ID");
                        String sku_price = jsonObject.getString("PRICE");
                        String tm_id = jsonObject.getString("TM_ID");
                        String category3_id = jsonObject.getString("CATEGORY3_ID");
                        String sku_create_time = jsonObject.getString("CREATE_TIME");
                        value.setSpu_id(spu_id);
                        value.setSku_price(sku_price);
                        value.setTm_id(tm_id);
                        value.setCategory3_id(category3_id);
                        value.setSku_create_time(sku_create_time);
                    }

                }, 60 * 5, TimeUnit.SECONDS);

        //TODO 6.5. 通过异步IO的方式关联spu维度数据
        SingleOutputStreamOperator<OrderWide> orderWideDSWithSpu = AsyncDataStream.unorderedWait(orderWideDSWithSku,
                new DimAsyncFunction<OrderWide>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2<String, String>[] tuple2s = new Tuple2[1];
                        tuple2s[0] = Tuple2.of("spu_id", "id");
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_spu_info";
                    }

                    @Override
                    public void join(OrderWide value, JSONObject jsonObject) {
                        String spu_name = jsonObject.getString("SPU_NAME");
                        value.setSpu_name(spu_name);
                    }
                }, 60 * 5, TimeUnit.SECONDS);

        //TODO 6.6. 通过异步IO的方式关联trademark维度数据
        SingleOutputStreamOperator<OrderWide> orderWideDSWithTrademark = AsyncDataStream.unorderedWait(orderWideDSWithSpu,
                new DimAsyncFunction<OrderWide>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2<String, String>[] tuple2s = new Tuple2[1];
                        tuple2s[0] = Tuple2.of("tm_id", "id");
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_base_trademark";
                    }

                    @Override
                    public void join(OrderWide value, JSONObject jsonObject) {
                        String tm_name = jsonObject.getString("TM_NAME");
                        value.setTm_name(tm_name);
                    }
                }, 60 * 5, TimeUnit.SECONDS);

        //TODO 6.7. 通过异步IO的方式关联category3维度数据
        SingleOutputStreamOperator<OrderWide> orderWideDSWithC3 = AsyncDataStream.unorderedWait(orderWideDSWithTrademark,
                new DimAsyncFunction<OrderWide>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2<String, String>[] tuple2s = new Tuple2[1];
                        tuple2s[0] = Tuple2.of("category3_id", "id");
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_base_category3";
                    }

                    @Override
                    public void join(OrderWide value, JSONObject jsonObject) {
                        String category3_name = jsonObject.getString("NAME");
                        String category2_id = jsonObject.getString("CATEGORY2_ID");
                        value.setCategory3_name(category3_name);
                        value.setCategory2_id(category2_id);
                    }
                }, 60 * 5, TimeUnit.SECONDS);

        //TODO 6.8. 通过异步IO的方式关联category3维度数据
        SingleOutputStreamOperator<OrderWide> orderWideDSWithC2 = AsyncDataStream.unorderedWait(orderWideDSWithC3,
                new DimAsyncFunction<OrderWide>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2<String, String>[] tuple2s = new Tuple2[1];
                        tuple2s[0] = Tuple2.of("category2_id", "id");
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_base_category2";
                    }

                    @Override
                    public void join(OrderWide value, JSONObject jsonObject) {
                        String category2_name = jsonObject.getString("NAME");
                        String category1_id = jsonObject.getString("CATEGORY1_ID");
                        value.setCategory2_name(category2_name);
                        value.setCategory1_id(category1_id);
                    }
                }, 60 * 5, TimeUnit.SECONDS);

        //TODO 6.9. 通过异步IO的方式关联category3维度数据
        SingleOutputStreamOperator<OrderWide> orderWideDSWithC1 = AsyncDataStream.unorderedWait(orderWideDSWithC2,
                new DimAsyncFunction<OrderWide>() {
                    @Override
                    public Tuple2<String, String>[] getConditions() {
                        Tuple2<String, String>[] tuple2s = new Tuple2[1];
                        tuple2s[0] = Tuple2.of("category1_id", "id");
                        return tuple2s;
                    }

                    @Override
                    public String getTable() {
                        return load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_base_category1";
                    }

                    @Override
                    public void join(OrderWide value, JSONObject jsonObject) {
                        String category1_name = jsonObject.getString("NAME");
                        value.setCategory1_name(category1_name);
                    }
                }, 60 * 5, TimeUnit.SECONDS);

        //TODO 7. 输出
        orderWideDSWithC1.print();
        orderWideDSWithC1.map(JSON::toJSONString).addSink(KafkaSinkUtil.getKafkaSink("DWM_ORDER_WIDE"));


        env.execute();


    }
}
